package com.huan.hadoop.serializable;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.IOException;

/**
 * map操作
 * 读取每行数据，将手机号(phone)作为key，具体的值(PhoneFlow)作为值
 *
 * @author huan.fu
 * @date 2023/7/10 - 19:29
 */
public class PhoneFlowMapper extends Mapper<LongWritable, Text, PhoneFlow, PhoneFlow> {

    private final PhoneFlow outPhoneKey = new PhoneFlow();
    private final PhoneFlow outValue = new PhoneFlow();

    @Override
    protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, PhoneFlow, PhoneFlow>.Context context) throws IOException, InterruptedException {
        // 获取一行数据 1	13736230513	192.196.100.1	www.baidu.com	2481	24681	200
        String line = value.toString();
        // 根据分隔符进行切割 [1,13736230513,192.196.100.1,www.baidu.com,2481,24681,200]
        String[] cells = line.split("\\s+");

        // 手机号作为outKey
        String phone = cells[1];
        outPhoneKey.setPhone(phone);

        // 封装outValue
        outValue.setPhone(phone);
        outValue.setUpFlow(Long.parseLong(cells[4]));
        outValue.setDownFlow(Long.parseLong(cells[5]));
        outValue.setSumFlow(outValue.getUpFlow() + outValue.getDownFlow());

        // 写出 outKey outValue
        context.write(outPhoneKey, outValue);
    }
}
